package com.example.userservice.kafka;

import com.example.commom.enums.StatusCodeEnum;
import com.example.commom.exception.CustomException;
import com.example.commom.global.KafkaConstants;
import com.example.feign.entity.User;
import com.example.userservice.service.UserService;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;

/**
 *  专门用来处理 修改用户信息的 kafka 消费者
 */
@Slf4j
@Component
@Transactional
public class KafkaUpdateUserInfoReceiver {

    @Autowired
    private UserService userService;

    @Autowired
    private ObjectMapper objectMapper;

    @KafkaListener(topics = KafkaConstants.KAFKA_UPDATE_USER_INFO, groupId = KafkaConstants.KAFKA_GROUP_ID)
    public void receiveComment(String message, Acknowledgment acknowledgment) {

        log.info("收到消息: " + message);

        // 处理收到的书籍消息
        User user;
        try {
            user = objectMapper.readValue(message, User.class);

            // 插入数据库
            boolean flag = userService.saveOrUpdate(user);
            // 手动提交偏移量
            if (flag) {
                log.info(LocalDateTime.now() + ": 用户: " + user.getUserId() +
                        " 的信息修改完成!");
                acknowledgment.acknowledge();
            } else {
                throw new CustomException(StatusCodeEnum.FAILED.getMsg(), -1);
            }

        } catch (Exception e) {
            log.error(e.getMessage());
            throw new CustomException(e.getMessage(), StatusCodeEnum.FAILED.getCode());
        }

    }
}
